package com.yinanxi.cool

import com.yinanxi.cool.beans.LogBean

import java.util.Date
import org.apache.commons.lang3.{RandomStringUtils, StringUtils}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}


/**
 * @Created by Orion
 * @Description
 * 计算逻辑:    同一个session中两个事件之间的时间间隔 > 30分钟   进行session分割
 * 计算数据源:   ods.cool_app_log
 */
object SessionSpliter {
  def main(args: Array[String]): Unit = {
  /*  if (args.length <=0){
      var  dt = new Date()
    }*/

    System.setProperty("HADOOP_USER_NAME", "root")
    // 使用Sparksql  加载hive中的数据
    val session = SparkSession.builder()
      .master("local[*]")
      .appName("session-split")
      .enableHiveSupport()
      .getOrCreate()

    // 加载hive中的表   //日期
    val df = session.read.table("ods.cool_app_log").where("dt="+args(0))

    // 将DF转换成RDD  使用算子来处理数据
    val rdd: RDD[Row] = df.rdd
    //
    val logs = rdd.map(row => {
      // 解析数据
      val account = row.getAs[String]("account")
      val appid = row.getAs[String]("appid")
      val appversion = row.getAs[String]("appversion")
      val carrier = row.getAs[String]("carrier")
      val deviceid = row.getAs[String]("deviceid")
      val devicetype = row.getAs[String]("devicetype")
      val eventid = row.getAs[String]("eventid")
      val ip = row.getAs[String]("ip")

      val latitude = row.getAs[Double]("latitude")
      val longitude = row.getAs[Double]("longitude")

      val nettype = row.getAs[String]("nettype")
      val osname = row.getAs[String]("osname")
      val osversion = row.getAs[String]("osversion")
      val properties = row.getAs[Map[String, String]]("properties")
      val releasechannel = row.getAs[String]("releasechannel")
      val resolution = row.getAs[String]("resolution")
      val sessionid = row.getAs[String]("sessionid")
      val timestamp = row.getAs[Long]("timestamp")
      val dt = row.getAs[String]("dt")
      val log = LogBean(account, appid, appversion, carrier, deviceid, devicetype, eventid, ip, latitude, longitude, nettype, osname, osversion, properties, releasechannel, resolution, sessionid, timestamp, dt, "")
      log
    })

    // logs.take(10).foreach(println)
    // 对数据进行过滤  sessionid != null  sessionid != "  "
    val filtered = logs.filter(log => log.sessionid != null && log.sessionid.trim != "")
    // 按照sessionid分组  相同sessionid的数据在一起
    val grouped: RDD[(String, Iterable[LogBean])] = filtered.groupBy(_.sessionid)
    // 分割完毕,  生成新的sessionid  new_session
    val sessioned: RDD[LogBean] = grouped.flatMap(tp => {
      val sessionid = tp._1
      // 按照时间排序
      val sorted = tp._2.toList.sortBy(_.timestamp)
      // 下面一个事件和上面一个事件  之间的时间差值  是否在30m以内
      if (sorted != null && sorted.size > 0) {
        var new_session: String = RandomStringUtils.randomAlphabetic(10)
        for (index <- 0 until sorted.size) { //遍历集合中所有的元素
          try {
            val preLog = sorted(index)



            preLog.new_session = new_session // 所有的数据都会复制 newsession
            val afterLog = sorted(index + 1)
            if (index < sorted.size - 1 && (afterLog.timestamp - preLog.timestamp) > 30 * 60 * 1000) {
              // 超过了30分 , 更新session
              new_session = RandomStringUtils.randomAlphabetic(10)
            }
          } catch {
            case e: Exception => null
          }
        }
      }
      sorted
    })
    // 将切分session后的数据存储到hive中 输出hive中
    // 将RDD 转换成  DF
   // import session.implicits._
   // val resDF: DataFrame = sessioned.toDF
    sessioned.take(10).foreach(println)

    /**
     * DataFrame 可以将自身的数据直接写入到HIVE中
     * 指定 分区
     * 指定存储格式
     * 指定写入模式
     */
 //   resDF.write.format("orc").mode(SaveMode.Overwrite).partitionBy("dt").saveAsTable("tmp.cool_app_session_split")

    session.close()
  }
}
